package doit20.sparksql

import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}


/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-04-10
 * @desc 将dataframe写入hive中
 */
object Demo14 {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("dsl风格api演示")
      .config("spark.sql.shuffle.partitions",1)  // spark-submit --master yarn --deploy-mode cluster --spark.sql.shuffle.partitions=400
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .master("local")
      .enableHiveSupport()
      .getOrCreate()

    // id,name,age,score,gender
    val schema = StructType(Seq(
      StructField("id", DataTypes.IntegerType),
      StructField("name", DataTypes.StringType),
      StructField("age", DataTypes.IntegerType),
      StructField("score", DataTypes.DoubleType),
      StructField("gender", DataTypes.StringType)
    ))

    val frame = spark.read.schema(schema).option("header", "true").csv("data/stu2.txt")

    // 如果目标表不存在，则会创建表，并将文件格式设置为parquet
    // frame.write.saveAsTable("stu")


    // 如果目标表已存在，且文件格式是parquet，那也不行，写不进去，会说DataSource不匹配
    // frame.write.mode(SaveMode.Append).saveAsTable("stu_hive")

    // 需要指定format为hive这种数据源
    //frame.write.format("hive").mode(SaveMode.Append).saveAsTable("stu_hive")


    // 还有一种办法，不指定format
    // 但本方式有一个缺陷，它不会检查字段的对应关系，只按顺序写入
    frame.createTempView("tmp")
    // spark.sql(
    //   """
    //     |
    //     |insert into table stu_hive
    //     | select id,name,age,score,gender
    //     |from tmp
    //     |""".stripMargin)


    /**
     * 写分区表
     */
    // 表不存在，自动创建分区表
    // frame.write.partitionBy("gender").saveAsTable("stu_partition")

    // 表已存在，追加写入，注意，这里用的是hive的动态分区插入机制，需要关闭strictmode严格模式，在sparksql中可以通过config来设置
    frame.write
      .format("hive")
      .mode(SaveMode.Append)
      .partitionBy("gender")
      .saveAsTable("stu2_partition")




    spark.close()

  }


}
